package cn.itcast.flink.base;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Author itcast
 * Date 2021/7/27 17:00
 * 将本地数据通过 jdbc 连接器写入到 mysql 中
 */
public class JDBCConnector {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.Source
        DataStreamSource<Student> source = env.fromElements(new Student(null, "xiaomage", 38));
        //不做转换操作
        //将数据源保存到 mysql connector
        source.addSink(JdbcSink.sink(
                "INSERT INTO t_student(`id`,`name`,`age`) values(null,?,?)",
                (ps, s) -> {
                    ps.setString(1,s.name);
                    ps.setInt(2,s.age);
                },
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                .withDriverName("com.mysql.jdbc.Driver")
                .withUsername("root")
                .withPassword("123456")
                .withUrl("jdbc:mysql://node3:3306/bigdata?useSSL=false")
                .build()
        ));
        env.execute();
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Student {
        private Integer id;
        private String name;
        private Integer age;
    }
}
